package com.innovation.ic.cc.base.thread.listener.rabbitmq;

import com.alibaba.fastjson.JSONObject;
import com.innovation.ic.cc.base.pojo.constant.handler.RabbitMqConstants;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

public class AbstractRabbitmqThread {

    private Logger logger = LoggerFactory.getLogger(this.getClass());

    /** 操作类型 */
    protected String type;

    /** 数据库变动内容 */
    protected ConsumerRecord<?, ?> consumer;

    /** 操作数据 */
    protected JSONObject dataJsonObject;



    /**
     * 获取消息体中的数据内容
     * @param bodyString 消息体
     * @return 返回消息体中的数据内容
     */
    protected String getMessageBody(String bodyString) {
        String messageBody = null;
        JSONObject json = (JSONObject) JSONObject.parse(bodyString);
        if (json != null && !json.isEmpty()) {
            messageBody = json.getString(RabbitMqConstants.MESSAGE_BODY);
        }
        return messageBody;
    }

    /**
     * 打印监听到的数据库变动内容
     * @param consumer consumer
     */
    protected String printConsumerContent(ConsumerRecord<?, ?> consumer) {
        logger.info("topic【" + consumer.topic() + "】，key【" + consumer.key() + "】，" +
                "分区位置【" + consumer.partition() + "】，下标【" + consumer.offset() + "】，" +
                "value【" + consumer.value() + "】");
        String value = (String) consumer.value();
        JSONObject valueJSONObject = JSONObject.parseObject(value);
        String zookeeperNode = valueJSONObject.getJSONArray("data").getJSONObject(0).getString("zookeeper_node");
        return zookeeperNode;
    }

}
